package com.boyi.shared.task;

import com.boyi.shared.util.RandomUtil;
import lombok.extern.log4j.Log4j2;

import java.util.ArrayList;
import java.util.Collection;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ForkJoinPool;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.RecursiveAction;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * @author tingwen
 * @description 批处理引擎
 * @date 2019-07-26 10:49
 **/
@Log4j2
public class JobEngine<T> {

//    private final static ForkJoinPool forkJoinPool = (ForkJoinPool) Executors.newWorkStealingPool();
    private final static ForkJoinPool forkJoinPool = (ForkJoinPool) new ForkJoinPool
        (8,
                ForkJoinPool.defaultForkJoinWorkerThreadFactory,
                null, true);
    /**
     * 可用线程
     */
//    private final static int taskNum = Runtime.getRuntime().availableProcessors();
    private final static int taskNum = 8;
    /**
     * 业务操作
     */
    private ITaskProcessor<T> myTask;
    private T[] data;
    //工作的任务个数
    private final int jobLength;
    //成功处理的任务数
    private final AtomicInteger successCount;
    //已处理的任务数
    private final AtomicInteger taskProcesserCount;
    //结果队列，拿结果从头拿，放结果从尾部放
    private final LinkedBlockingDeque<TaskResult> taskDetailQueue;
    private final String name;

    public JobEngine(ITaskProcessor<T> myTask, Collection<T> collection) {
        this(myTask, (T[]) collection.toArray());
    }

    public JobEngine(ITaskProcessor<T> myTask, T[] data) {
        this.name = "任务" + RandomUtil.getRandom(0, 100);
        this.myTask = myTask;
        this.data = data;
        jobLength = data.length;
        this.successCount = new AtomicInteger(0);
        this.taskProcesserCount = new AtomicInteger(0);
        this.taskDetailQueue = new LinkedBlockingDeque<TaskResult>(jobLength);
    }

    private ForkAndJoinRequest task;
    public volatile boolean flag = true;

    /**
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void start() throws ExecutionException, InterruptedException {
        task = new ForkAndJoinRequest(0, data.length - 1, data.length / taskNum);
        forkJoinPool.execute(task);
        task.join();
        task.get();
        taskDetailQueue.clear();
        log.info(name + " 执行完毕");
    }

    /**
     * @throws ExecutionException
     * @throws InterruptedException
     */
    public void stop() throws ExecutionException, InterruptedException {
        flag = false;
        taskDetailQueue.clear();
        log.info(name + " 强制停止");
    }

    //返回成功处理的结果数
    public int getSuccessCount() {
        return successCount.get();
    }

    //返回当前已处理的结果数
    public int getTaskProcesserCount() {
        return taskProcesserCount.get();
    }

    //提供工作中失败的次数，课堂上没有加，只是为了方便调用者使用
    public int getFailCount() {
        return taskProcesserCount.get() - successCount.get();
    }

    public String getTotalProcess() {
        return "Success[" + successCount.get() + "]/Current["
                + taskProcesserCount.get() + "] Total[" + jobLength + "]";
    }

    //获得工作中每个任务的处理详情
    public List<TaskResult> getTaskDetail() {
        List<TaskResult> taskList = new LinkedList<>();
        TaskResult taskResult;
        //从阻塞队列中拿任务的结果，反复取，一直取到null为止，说明目前队列中最新的任务结果已经取完，可以不取了
        while ((taskResult = taskDetailQueue.pollFirst()) != null) {
            taskList.add(taskResult);
        }
        return taskList;
    }

    //放任务的结果，从业务应用角度来说，保证最终一致性即可，不需要对方法加锁.
    public void addTaskResult(TaskResult result) {
        if (TaskResultType.Success.equals(result.getResultType())) {
            successCount.incrementAndGet();
        }
//        taskDetailQueue.addLast(result);
        taskProcesserCount.incrementAndGet();
    }


    private class ForkAndJoinRequest extends RecursiveAction {
        private int start;
        private int end;
        private int count;

        public ForkAndJoinRequest(int start, int end, int count) {
            this.start = start;
            this.end = end;
            this.count = count;
        }

        @Override
        protected void compute() {
            List<ForkAndJoinRequest> subTasks = new ArrayList<>();
            int num = end - start;
            if (num <= count) {
                log.info("start = " + start + " end = " + end + " count = " + count);
                for (int i = start; i <= end; i++) {
//                    myTask.taskExecute(data[i]);
                    if (flag) {
                        TaskResult result = null;
                        try {
                            result = myTask.taskSubmit(data[i]);
                            //要做检查，防止开发人员处理不当
                            if (result == null) {
                                result = new TaskResult(TaskResultType.Exception, null, "result is null");
                            }
                            if (result.getResultType() == null) {
                                if (result.getReason() == null) {
                                    result = new TaskResult(TaskResultType.Exception, result.getReturnValue(), "reason is null");
                                } else {
                                    result = new TaskResult(TaskResultType.Exception, result.getReturnValue(), "result is null,but reason:" + result.getReason());
                                }
                            }
                        } catch (Exception e) {
                            e.printStackTrace();
                            result = new TaskResult(TaskResultType.Exception, null, e.getMessage());
                        } finally {
//                            log.info(getTotalProcess());
                            addTaskResult(result);
                        }
                    }
                }
            } else { //否则fork出其他request
                int mid = (start + end) / 2;
                ForkAndJoinRequest request1 = new ForkAndJoinRequest(start, mid, count);
                //调用fork方法将自身放入等待队列中等待执行
                ForkAndJoinRequest request2 = new ForkAndJoinRequest(mid + 1, end, count);
                subTasks.add(request1);
                subTasks.add(request2);
            }
            if (!subTasks.isEmpty()) {
                for (ForkAndJoinRequest subTask : invokeAll(subTasks)) {
                    subTask.join();//等待子任务执行完成
                }
            }
        }

    }

}
